package demo.core;

import demo.message.InMemoryMessageHandler;
import demo.message.RaftMessage;
import demo.message.RaftMessageHandler;
import demo.storage.LogEntry;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Raft节点实现
 *
 * <p>这是Raft分布式共识算法的核心实现，每个RaftNode代表集群中的一个节点。
 * 该实现完整支持Raft算法的三个关键子问题：</p>
 *
 * <ul>
 *   <li><b>Leader Election（领导者选举）</b>：自动选举和维护集群领导者</li>
 *   <li><b>Log Replication（日志复制）</b>：Leader将客户端请求复制到集群</li>
 *   <li><b>Safety（安全性）</b>：确保在任何情况下都不违反一致性</li>
 * </ul>
 *
 * <p><b>核心特性：</b></p>
 * <ul>
 *   <li>强一致性：所有节点最终会达到相同的状态</li>
 *   <li>高可用性：只要多数节点存活，集群就能正常工作</li>
 *   <li>故障恢复：节点可以随时加入或离开集群</li>
 *   <li>随机化选举：避免选举冲突，确保收敛</li>
 * </ul>
 *
 * <p><b>线程安全：</b></p>
 * <p>本实现使用synchronized关键字和并发安全的数据结构来保证线程安全，
 * 所有公共方法都可以在多线程环境中安全调用。</p>
 *
 * <p><b>状态持久化：</b></p>
 * <p>当前实现将状态保存在内存中，实际生产环境需要将关键状态持久化到磁盘。
 * 需要持久化的状态包括：currentTerm、votedFor和log。</p>
 *
 * @author Raft Implementation Team
 * @version 1.0
 * @since 1.0
 */
public class RaftNode {

    // ==================== 基本配置 ====================

    /**
     * 节点唯一标识符
     */
    private final String nodeId;

    /**
     * 集群中其他节点的ID列表
     */
    private final List<String> peers;

    /**
     * 消息处理器，负责节点间通信
     */
    private final RaftMessageHandler messageHandler;

    // ==================== 持久化状态 ====================
    // 这些状态在所有节点重启后必须保持不变

    /**
     * 当前任期号，单调递增
     */
    private volatile int currentTerm = 0;

    /**
     * 当前任期内收到投票的候选者ID（如果没有则为null）
     */
    private volatile String votedFor = null;

    /**
     * 日志条目数组，索引从1开始（索引0为占位条目）
     */
    private final List<LogEntry> log = new ArrayList<>();

    // ==================== 易失性状态 ====================
    // 这些状态在重启时会丢失，可以从其他状态重新计算

    /**
     * 当前节点状态
     */
    private volatile RaftState state = RaftState.FOLLOWER;

    /**
     * 已知已提交的最高日志条目索引
     */
    private volatile int commitIndex = 0;

    /**
     * 已应用到状态机的最高日志条目索引
     */
    private volatile int lastApplied = 0;

    // ==================== 领导者状态 ====================
    // 只有Leader需要维护这些状态，选举后重新初始化

    /**
     * 对于每个服务器，下一个要发送的日志条目索引
     */
    private final Map<String, Integer> nextIndex = new ConcurrentHashMap<>();

    /**
     * 对于每个服务器，已知已复制的最高日志条目索引
     */
    private final Map<String, Integer> matchIndex = new ConcurrentHashMap<>();

    // ==================== 选举和心跳控制 ====================

    /**
     * 定时任务调度器，管理选举超时和心跳
     */
    private volatile ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(3);

    /**
     * 最后一次收到Leader心跳的时间戳
     */
    private volatile long lastHeartbeat = System.currentTimeMillis();

    /**
     * 选举是否正在进行中（避免重复选举）
     */
    private final AtomicBoolean electionInProgress = new AtomicBoolean(false);

    /**
     * 当前选举周期收到的投票数
     */
    private final AtomicInteger votesReceived = new AtomicInteger(0);

    // ==================== 配置参数 ====================

    /**
     * 选举超时时间下限（毫秒）
     */
    private static final int ELECTION_TIMEOUT_MIN = 150;

    /**
     * 选举超时时间上限（毫秒）
     */
    private static final int ELECTION_TIMEOUT_MAX = 300;

    /**
     * Leader心跳间隔（毫秒）
     */
    private static final int HEARTBEAT_INTERVAL = 50;

    /**
     * 构造Raft节点
     *
     * <p>创建一个新的Raft节点并初始化所有必要的状态。节点创建后会：</p>
     * <ul>
     *   <li>以FOLLOWER状态启动</li>
     *   <li>创建空的日志（索引0作为占位符）</li>
     *   <li>启动选举超时定时器</li>
     *   <li>启动日志应用定时器</li>
     * </ul>
     *
     * <p><b>注意：</b>创建节点后，需要将其注册到MessageHandler中，
     * 以便其他节点能够与其通信。</p>
     *
     * @param nodeId         节点的唯一标识符，不能为null或空
     * @param peers          集群中其他节点的ID列表，不能为null（可以为空，表示单节点集群）
     * @param messageHandler 消息处理器，负责节点间通信，不能为null
     * @throws IllegalArgumentException 如果任何参数不合法
     */
    public RaftNode(String nodeId, List<String> peers, RaftMessageHandler messageHandler) {
        // 参数验证
        if (nodeId == null || nodeId.trim().isEmpty()) {
            throw new IllegalArgumentException("节点ID不能为空");
        }
        if (peers == null) {
            throw new IllegalArgumentException("节点列表不能为null");
        }
        if (messageHandler == null) {
            throw new IllegalArgumentException("消息处理器不能为null");
        }

        this.nodeId = nodeId;
        this.peers = new ArrayList<>(peers); // 防御性复制
        this.messageHandler = messageHandler;

        // 初始化日志，索引从1开始，索引0为占位条目
        log.add(new LogEntry(0, 0, ""));

        // 启动定时器
        startElectionTimer();    // 选举超时检测
        startApplyLogTimer();    // 日志应用到状态机

        System.out.println("Raft节点 " + nodeId + " 已启动，集群大小: " + (peers.size() + 1));
    }

    /**
     * 处理投票请求
     */
    public synchronized RaftMessage.VoteResponse handleVoteRequest(RaftMessage.VoteRequest request) {
        System.out.println(String.format("[%s] 收到来自节点 %s 的投票请求: {term=%d, lastLogIndex=%d, lastLogTerm=%d}",
            nodeId, request.getCandidateId(), request.getTerm(), request.getLastLogIndex(), request.getLastLogTerm()));

        // 如果请求任期小于当前任期，拒绝投票
        if (request.getTerm() < currentTerm) {
            System.out.println(String.format("[%s] 拒绝投票：请求任期(%d) < 当前任期(%d)",
                nodeId, request.getTerm(), currentTerm));
            return new RaftMessage.VoteResponse(currentTerm, nodeId, request.getCandidateId(), false);
        }

        // 如果请求任期大于当前任期，更新当前任期并转为Follower
        if (request.getTerm() > currentTerm) {
            System.out.println(String.format("[%s] 发现更高任期：%d -> %d，转为Follower",
                nodeId, currentTerm, request.getTerm()));
            currentTerm = request.getTerm();
            state = RaftState.FOLLOWER;
            votedFor = null;
        }

        // 检查是否已经投票给其他候选者
        if (votedFor != null && !votedFor.equals(request.getCandidateId())) {
            System.out.println(String.format("[%s] 拒绝投票：已经投票给节点 %s",
                nodeId, votedFor));
            return new RaftMessage.VoteResponse(currentTerm, nodeId, request.getCandidateId(), false);
        }

        // 检查候选者日志是否至少和自己一样新
        int lastLogIndex = log.size() - 1;
        int lastLogTerm = lastLogIndex > 0 ? log.get(lastLogIndex).getTerm() : 0;
        
        if (request.getLastLogTerm() < lastLogTerm || 
            (request.getLastLogTerm() == lastLogTerm && request.getLastLogIndex() < lastLogIndex)) {
            System.out.println(String.format("[%s] 拒绝投票：候选者日志(%d:%d)不如本地日志(%d:%d)新",
                nodeId, request.getLastLogIndex(), request.getLastLogTerm(), lastLogIndex, lastLogTerm));
            return new RaftMessage.VoteResponse(currentTerm, nodeId, request.getCandidateId(), false);
        }

        // 投票给候选者
        votedFor = request.getCandidateId();
        System.out.println(String.format("[%s] 投票给节点 %s，任期 %d",
            nodeId, request.getCandidateId(), currentTerm));
        
        // 重置选举定时器
        lastHeartbeat = System.currentTimeMillis();
        
        return new RaftMessage.VoteResponse(currentTerm, nodeId, request.getCandidateId(), true);
    }

    /**
     * 处理附加日志请求
     */
    public synchronized RaftMessage.AppendEntriesResponse handleAppendEntries(RaftMessage.AppendEntriesRequest request) {
        System.out.println(String.format("[%s] 收到来自Leader %s 的附加日志请求: {term=%d, prevLogIndex=%d, prevLogTerm=%d, entries.size=%d, leaderCommit=%d}",
            nodeId, request.getLeaderId(), request.getTerm(), request.getPrevLogIndex(), request.getPrevLogTerm(),
            request.getEntries() != null ? request.getEntries().size() : 0, request.getLeaderCommit()));

        // 如果请求任期小于当前任期，拒绝请求
        if (request.getTerm() < currentTerm) {
            System.out.println(String.format("[%s] 拒绝附加日志：请求任期(%d) < 当前任期(%d)",
                nodeId, request.getTerm(), currentTerm));
            return new RaftMessage.AppendEntriesResponse(currentTerm, nodeId, request.getLeaderId(), false, log.size() - 1);
        }

        // 重置选举定时器
        lastHeartbeat = System.currentTimeMillis();

        // 如果请求任期大于当前任期，更新当前任期
        if (request.getTerm() > currentTerm) {
            System.out.println(String.format("[%s] 发现更高任期：%d -> %d",
                nodeId, currentTerm, request.getTerm()));
            currentTerm = request.getTerm();
            votedFor = null;
        }

        // 转为Follower（如果不是的话）
        if (state != RaftState.FOLLOWER) {
            System.out.println(String.format("[%s] 状态转换：%s -> FOLLOWER",
                nodeId, state));
            state = RaftState.FOLLOWER;
        }

        // 验证前一个日志条目
        if (request.getPrevLogIndex() >= log.size() ||
            (request.getPrevLogIndex() > 0 && log.get(request.getPrevLogIndex()).getTerm() != request.getPrevLogTerm())) {
            System.out.println(String.format("[%s] 日志一致性检查失败：prevLogIndex=%d, localLogSize=%d",
                nodeId, request.getPrevLogIndex(), log.size()));
            return new RaftMessage.AppendEntriesResponse(currentTerm, nodeId, request.getLeaderId(), false, log.size() - 1);
        }

        // 追加新的日志条目
        if (request.getEntries() != null && !request.getEntries().isEmpty()) {
            // 删除冲突的旧日志条目
            int newEntriesIndex = 0;
            for (int i = request.getPrevLogIndex() + 1; i < log.size() && newEntriesIndex < request.getEntries().size(); i++) {
                if (log.get(i).getTerm() != request.getEntries().get(newEntriesIndex).getTerm()) {
                    System.out.println(String.format("[%s] 删除冲突日志：从索引 %d 开始",
                        nodeId, i));
                    while (log.size() > i) {
                        log.remove(log.size() - 1);
                    }
                    break;
                }
                newEntriesIndex++;
            }

            // 追加新的日志条目
            for (; newEntriesIndex < request.getEntries().size(); newEntriesIndex++) {
                LogEntry entry = request.getEntries().get(newEntriesIndex);
                if (log.size() > request.getPrevLogIndex() + 1 + newEntriesIndex) {
                    log.set(request.getPrevLogIndex() + 1 + newEntriesIndex, entry);
                } else {
                    log.add(entry);
                }
                System.out.println(String.format("[%s] 追加日志：索引=%d, 任期=%d, 命令=%s",
                    nodeId, entry.getIndex(), entry.getTerm(), entry.getCommand()));
            }
        }

        // 更新commitIndex
        if (request.getLeaderCommit() > commitIndex) {
            int oldCommitIndex = commitIndex;
            commitIndex = Math.min(request.getLeaderCommit(), log.size() - 1);
            System.out.println(String.format("[%s] 更新commitIndex：%d -> %d",
                nodeId, oldCommitIndex, commitIndex));
        }

        return new RaftMessage.AppendEntriesResponse(currentTerm, nodeId, request.getLeaderId(), true, log.size() - 1);
    }

    /**
     * 开始选举
     */
    private synchronized void startElection() {
        if (!electionInProgress.compareAndSet(false, true)) {
            System.out.println(String.format("[%s] 已有选举正在进行中，跳过本次选举",
                nodeId));
            return;
        }

        try {
            // 转换为候选者状态
            state = RaftState.CANDIDATE;
            currentTerm++;
            votedFor = nodeId;
            votesReceived.set(1); // 给自己投票

            System.out.println(String.format("[%s] 开始选举：任期=%d, 日志大小=%d",
                nodeId, currentTerm, log.size()));

            // 获取最后一条日志的信息
            int lastLogIndex = log.size() - 1;
            int lastLogTerm = lastLogIndex > 0 ? log.get(lastLogIndex).getTerm() : 0;

            // 向其他节点发送投票请求
            for (String peer : peers) {
                CompletableFuture.runAsync(() -> {
                    try {
                        RaftMessage.VoteRequest request = new RaftMessage.VoteRequest(
                            currentTerm, nodeId, nodeId, lastLogIndex, lastLogTerm);
                        
                        System.out.println(String.format("[%s] 向节点 %s 发送投票请求",
                            nodeId, peer));
                        
                        RaftMessage.VoteResponse response = messageHandler.sendVoteRequest(peer, request);
                        handleVoteResponse(response);
                    } catch (Exception e) {
                        System.err.println(String.format("[%s] 向节点 %s 发送投票请求失败: %s",
                            nodeId, peer, e.getMessage()));
                    }
                });
            }
        } finally {
            electionInProgress.set(false);
        }
    }

    /**
     * 处理投票响应
     */
    private synchronized void handleVoteResponse(RaftMessage.VoteResponse response) {
        System.out.println(String.format("[%s] 收到投票响应：来自=%s, 任期=%d, 同意=%b",
            nodeId, response.getSenderId(), response.getTerm(), response.isGranted()));

        // 如果不再是候选者，忽略响应
        if (state != RaftState.CANDIDATE) {
            System.out.println(String.format("[%s] 已不是候选者（当前状态=%s），忽略投票响应",
                nodeId, state));
            return;
        }

        // 如果发现更高的任期，转为Follower
        if (response.getTerm() > currentTerm) {
            System.out.println(String.format("[%s] 发现更高任期：%d -> %d，转为Follower",
                nodeId, currentTerm, response.getTerm()));
            currentTerm = response.getTerm();
            state = RaftState.FOLLOWER;
            votedFor = null;
            return;
        }

        // 如果获得投票，增加计数
        if (response.isGranted()) {
            int votes = votesReceived.incrementAndGet();
            System.out.println(String.format("[%s] 获得投票：当前票数=%d, 所需票数=%d",
                nodeId, votes, (peers.size() + 1) / 2 + 1));

            // 如果获得多数票，成为Leader
            if (votes >= (peers.size() + 1) / 2 + 1) {
                System.out.println(String.format("[%s] 获得多数票（%d票），成为Leader",
                    nodeId, votes));
                becomeLeader();
            }
        }
    }

    /**
     * 转换为Leader状态
     */
    private synchronized void becomeLeader() {
        if (state != RaftState.CANDIDATE) {
            return;
        }

        state = RaftState.LEADER;
        System.out.println(String.format("[%s] 成为Leader：任期=%d",
            nodeId, currentTerm));

        // 初始化Leader状态
        for (String peer : peers) {
            nextIndex.put(peer, log.size());
            matchIndex.put(peer, 0);
        }

        // 立即发送心跳
        sendHeartbeat();

        // 启动心跳定时器
        startHeartbeatTimer();
    }

    /**
     * 发送心跳
     */
    private void sendHeartbeat() {
        for (String peer : peers) {
            CompletableFuture.runAsync(() -> {
                try {
                    sendAppendEntries(peer);
                } catch (Exception e) {
                    System.err.println(String.format("[%s] 向节点 %s 发送心跳失败: %s",
                        nodeId, peer, e.getMessage()));
                }
            });
        }
    }

    /**
     * 发送附加日志请求
     */
    private void sendAppendEntries(String peer) {
        if (state != RaftState.LEADER) {
            return;
        }

        synchronized (this) {
            int prevLogIndex = nextIndex.get(peer) - 1;
            int prevLogTerm = log.get(prevLogIndex).getTerm();

            List<LogEntry> entries = new ArrayList<>();
            for (int i = nextIndex.get(peer); i < log.size(); i++) {
                entries.add(log.get(i));
            }

            RaftMessage.AppendEntriesRequest request = new RaftMessage.AppendEntriesRequest(
                    currentTerm,
                    nodeId,          // senderId
                    nodeId,          // leaderId (发送者就是leader)
                    prevLogIndex,    // prevLogIndex
                    prevLogTerm,     // prevLogTerm
                    entries,         // entries
                    commitIndex      // leaderCommit
            );

            try {
                RaftMessage.AppendEntriesResponse response = messageHandler.sendAppendEntries(peer, request);
                handleAppendEntriesResponse(peer, response, entries.size());
            } catch (Exception e) {
                System.err.println("发送附加日志到 " + peer + " 失败: " + e.getMessage());
            }
        }
    }

    /**
     * 处理附加日志响应
     */
    private void handleAppendEntriesResponse(String peer, RaftMessage.AppendEntriesResponse response, int entriesSent) {
        synchronized (this) {
            if (state != RaftState.LEADER || response.getTerm() != currentTerm) {
                return;
            }

            // 如果响应的任期更大，转为跟随者
            if (response.getTerm() > currentTerm) {
                currentTerm = response.getTerm();
                state = RaftState.FOLLOWER;
                votedFor = null;
                return;
            }

            if (response.isSuccess()) {
                // 更新索引
                nextIndex.put(peer, nextIndex.get(peer) + entriesSent);
                matchIndex.put(peer, nextIndex.get(peer) - 1);

                // 更新提交索引
                updateCommitIndex();
            } else {
                // 日志不一致，回退到跟随者的最后日志索引
                int lastLogIndex = response.getLastLogIndex();
                nextIndex.put(peer, Math.max(1, lastLogIndex + 1));
            }
        }
    }

    /**
     * 更新提交索引
     */
    private void updateCommitIndex() {
        for (int n = commitIndex + 1; n < log.size(); n++) {
            if (log.get(n).getTerm() == currentTerm) {
                int count = 1; // 自己
                for (String peer : peers) {
                    if (matchIndex.get(peer) >= n) {
                        count++;
                    }
                }

                if (count > (peers.size() + 1) / 2) {
                    commitIndex = n;
                    System.out.println("节点 " + nodeId + " 提交索引更新为 " + commitIndex);
                }
            }
        }
    }

    /**
     * 添加日志条目（仅领导者）
     */
    public boolean appendEntry(String command) {
        if (state != RaftState.LEADER) {
            return false;
        }

        synchronized (this) {
            LogEntry entry = new LogEntry(currentTerm, log.size(), command);
            log.add(entry);
            System.out.println("领导者 " + nodeId + " 添加日志: " + command);

            // 立即尝试复制到跟随者
            sendHeartbeat();
            return true;
        }
    }

    /**
     * 启动选举定时器
     */
    private void startElectionTimer() {
        scheduler.scheduleWithFixedDelay(() -> {
            long timeSinceLastHeartbeat = System.currentTimeMillis() - lastHeartbeat;
            int currentTimeout = getRandomElectionTimeout();

            if (timeSinceLastHeartbeat > currentTimeout && state != RaftState.LEADER) {
                System.out.println("节点 " + nodeId + " 选举超时(" + currentTimeout + "ms)，开始新选举");
                startElection();
            }
        }, ELECTION_TIMEOUT_MIN, 50, TimeUnit.MILLISECONDS);
    }

    /**
     * 启动心跳定时器
     */
    private void startHeartbeatTimer() {
        scheduler.scheduleWithFixedDelay(() -> {
            if (state == RaftState.LEADER) {
                sendHeartbeat();
            }
        }, 0, HEARTBEAT_INTERVAL, TimeUnit.MILLISECONDS);
    }

    /**
     * 启动应用日志定时器
     */
    private void startApplyLogTimer() {
        scheduler.scheduleWithFixedDelay(this::applyCommittedEntries, 0, 10, TimeUnit.MILLISECONDS);
    }

    /**
     * 应用已提交的日志条目
     */
    private void applyCommittedEntries() {
        synchronized (this) {
            while (lastApplied < commitIndex) {
                lastApplied++;
                LogEntry entry = log.get(lastApplied);
                if (!entry.getCommand().isEmpty()) {
                    System.out.println("节点 " + nodeId + " 应用日志: " + entry.getCommand());
                    // 这里可以添加状态机应用逻辑
                }
            }
        }
    }

    /**
     * 获取随机选举超时时间
     */
    private int getRandomElectionTimeout() {
        // 使用节点ID的哈希码作为随机种子的一部分，确保每个节点的随机性不同
        Random random = new Random(System.nanoTime() + nodeId.hashCode());
        int timeout = ELECTION_TIMEOUT_MIN + random.nextInt(ELECTION_TIMEOUT_MAX - ELECTION_TIMEOUT_MIN);

        // 记录随机超时时间，帮助观察随机性
        if (System.getProperty("raft.debug", "false").equals("true")) {
            System.out.println("节点 " + nodeId + " 设置随机选举超时: " + timeout + "ms");
        }

        return timeout;
    }

    /**
     * 获取节点状态信息
     */
    public String getStatus() {
        return String.format("节点[%s] 状态:%s 任期:%d 日志大小:%d 提交索引:%d",
                nodeId, state, currentTerm, log.size() - 1, commitIndex);
    }

    public String getNodeId() {
        return nodeId;
    }

    public RaftState getState() {
        return state;
    }

    public int getCurrentTerm() {
        return currentTerm;
    }

    public int getLogSize() {
        return log.size() - 1;
    }

    public int getCommitIndex() {
        return commitIndex;
    }

    /**
     * 关闭节点
     *
     * <p>立即关闭调度器，停止所有定时任务。这是一个快速关闭方法，
     * 不等待正在执行的任务完成。</p>
     *
     * <p>如果需要优雅关闭，请使用 {@link #gracefulShutdown()} 方法。</p>
     */
    public void shutdown() {
        synchronized (this) {
            if (scheduler != null && !scheduler.isShutdown()) {
                scheduler.shutdownNow(); // 立即中断所有任务
                System.out.println("节点 " + nodeId + " 已关闭（立即模式）");
            } else {
                System.out.println("节点 " + nodeId + " 调度器已关闭，无需再次关闭");
            }
        }
    }

    /**
     * 优雅关闭节点
     *
     * <p>优雅地关闭调度器，等待正在执行的任务完成，但不接受新任务。
     * 此方法会等待最多10秒让任务完成，超时后强制关闭。</p>
     *
     * @return true如果所有任务在超时前完成，false如果超时
     */
    public boolean gracefulShutdown() {
        synchronized (this) {
            if (scheduler == null || scheduler.isShutdown()) {
                System.out.println("节点 " + nodeId + " 调度器已关闭");
                return true;
            }

            System.out.println("节点 " + nodeId + " 开始优雅关闭...");
            scheduler.shutdown(); // 不接受新任务

            try {
                // 等待最多10秒让现有任务完成
                if (scheduler.awaitTermination(10, TimeUnit.SECONDS)) {
                    System.out.println("节点 " + nodeId + " 优雅关闭完成");
                    return true;
                } else {
                    System.out.println("节点 " + nodeId + " 优雅关闭超时，强制关闭");
                    scheduler.shutdownNow();
                    return false;
                }
            } catch (InterruptedException e) {
                System.out.println("节点 " + nodeId + " 关闭过程被中断，强制关闭");
                scheduler.shutdownNow();
                Thread.currentThread().interrupt();
                return false;
            }
        }
    }

    /**
     * 重新创建定时任务调度器
     *
     * <p>当调度器被shutdown后，需要重新创建才能继续工作。
     * 此方法会安全地关闭旧的调度器并创建新的调度器。</p>
     *
     * <p><b>注意：</b>此方法应该在synchronized块中调用，确保线程安全。</p>
     */
    private void recreateScheduler() {
        // 确保旧的调度器完全关闭
        if (scheduler != null && !scheduler.isShutdown()) {
            scheduler.shutdown();
            try {
                // 等待调度器完全关闭，最多等待5秒
                if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                    scheduler.shutdownNow();
                }
            } catch (InterruptedException e) {
                scheduler.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }

        // 创建新的调度器
        scheduler = Executors.newScheduledThreadPool(3, r -> {
            Thread t = new Thread(r);
            t.setName("RaftNode-" + nodeId + "-Scheduler-" + t.getId());
            t.setDaemon(true); // 设置为守护线程
            return t;
        });

        System.out.println("节点 " + nodeId + " 重新创建了调度器");
    }

    /**
     * 重新加入集群
     *
     * <p>当故障节点恢复时调用此方法。此方法会：</p>
     * <ul>
     *   <li>重置节点状态为FOLLOWER</li>
     *   <li>清除投票记录</li>
     *   <li>重新注册到消息处理器</li>
     *   <li>重新创建调度器（如果已关闭）</li>
     *   <li>重新启动所有定时任务</li>
     * </ul>
     *
     * <p><b>线程安全：</b>使用synchronized保证原子性操作。</p>
     *
     * <p><b>使用场景：</b></p>
     * <ul>
     *   <li>节点从网络分区中恢复</li>
     *   <li>节点进程重启后重新加入集群</li>
     *   <li>故障转移测试中模拟节点恢复</li>
     * </ul>
     */
    public void rejoinCluster() {
        synchronized (this) {
            // 重置为跟随者状态
            state = RaftState.FOLLOWER;
            votedFor = null;
            lastHeartbeat = System.currentTimeMillis();
            electionInProgress.set(false);
            votesReceived.set(0);

            System.out.println("节点 " + nodeId + " 重新加入集群，当前任期: " + currentTerm);

            // 重新注册到消息处理器
            InMemoryMessageHandler.registerNode(nodeId, this);

            // 重新启动定时器（如果已关闭）
            if (scheduler.isShutdown()) {
                System.out.println("节点 " + nodeId + " 检测到调度器已关闭，正在重新创建...");
                recreateScheduler();

                // 重新启动所有定时任务
                startElectionTimer();
                startApplyLogTimer();

                System.out.println("节点 " + nodeId + " 成功重新启动所有定时任务");
            } else {
                System.out.println("节点 " + nodeId + " 调度器仍在运行，无需重新创建");
            }
        }
    }

    /**
     * 强制设置任期（仅用于测试场景）
     */
    public void forceSetTerm(int term) {
        synchronized (this) {
            this.currentTerm = term;
            this.votedFor = null;
            System.out.println("节点 " + nodeId + " 强制设置任期为: " + term);
        }
    }

    /**
     * 强制设置为Leader状态（仅用于测试脑裂场景）
     */
    public void forceSetLeader(int term) {
        synchronized (this) {
            this.currentTerm = term;
            this.state = RaftState.LEADER;
            this.votedFor = nodeId;

            // 初始化领导者状态
            int nextLogIndex = log.size();
            for (String peer : peers) {
                nextIndex.put(peer, nextLogIndex);
                matchIndex.put(peer, 0);
            }

            System.out.println("节点 " + nodeId + " 强制设置为Leader，任期: " + term + " (测试用)");

            // 开始发送心跳
            startHeartbeatTimer();
        }
    }
} 